package net.wlm.realtime.job;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.wlm.realtime.funcation.LoadProvinceDimMapFunction;
import net.wlm.realtime.funcation.RequestDicDimAsyncFunction;
import net.wlm.realtime.utils.DateTimeUtil;
import net.wlm.realtime.utils.DorisUtil;
import net.wlm.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 交易-下单明细-事务事实表
 *          从Kafka读取订单相关表数据和Hbase读取相关维表数据，进行关联JOIN，构建订单明细宽表数据，写入Doris表；
 */

public class JtpOrderDetailDorisDwdJob {
    public static void main(String[] args) throws Exception {
    // 1.执行环境
        StreamExecutionEnvironment  env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
    // 2.数据源
        DataStream<String> dataStream = KafkaUtil.consumerKafka(env, "topic-db");
    // 3.数据转换
        DataStream<String> dwdStream = handle(dataStream);
    // 4.数据保存
        DorisUtil.saveToDoris(dwdStream, "jtp_mall_doris", "dwd_order_detail");
    // 5.触发执行
        env.execute();
    }

    /**
     * 将订单数据关联，并且添加维度信息，形成订单明细宽表，最后写入Doris，基于物化视图实时数据聚合。
     *  1-数据流JOIN：Interval Join
     *  2-省份维度补全：预加载
     *  3-字典维度补全：异步IO
     */
    private static DataStream<String> handle(DataStream<String> dataStream) {
        // 数据流JOIN：Interval Join,订单明细数据流 关联 订单信息数据流【大数据流 关联 大数据流】
        DataStream<String> joinStream =  joinProcess(dataStream);
        // 省份维度补全：异步IO
        DataStream<String> provinceStream = joinStream.map(new LoadProvinceDimMapFunction());
        // 字典维度补全：异步IO
        SingleOutputStreamOperator<String> dwdStream = AsyncDataStream.orderedWait(
                provinceStream,
                new RequestDicDimAsyncFunction(),
                10000,
                TimeUnit.MILLISECONDS,
                100
        );
        return dwdStream;
    }

    private static DataStream<String> joinProcess(DataStream<String> dataStream) {
        // 订单明细数据流
        SingleOutputStreamOperator<String> detailStream = dataStream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                // 解析JSON数据
                JSONObject jsonObject = JSON.parseObject(s);
                // 过滤
                return "order_detail".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });
        // 订单数据流
        SingleOutputStreamOperator<String> infoStream = dataStream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                return "order_info".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });
        // 指定事件时间和WaterMark水位线
        SingleOutputStreamOperator<String> timeInfoStream = infoStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String s, long l) {
                                // 解析
                                String eventTime = JSON.parseObject(s).getJSONObject("operate_date")
                                        .getString("create_time");
                                return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                            }
                        })
        );
        SingleOutputStreamOperator<String> timeDetailStream = detailStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String s, long l) {
                                // 解析
                                String eventTime = JSON.parseObject(s).getJSONObject("operate_date")
                                        .getString("create_time");
                                return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                            }
                        })
        );
        // 数据流关联join，由于info -> detail，所以使用 Interval Join
        // Interval Join 要求俩个数据流为分组流，按照关联字段分组KeyBy
        SingleOutputStreamOperator<String> joinStream = timeInfoStream.keyBy(element -> JSON.parseObject(element).getJSONObject("operate_date").getString("id"))
                .intervalJoin(
                        timeDetailStream.keyBy(
                                element -> JSON.parseObject(element).getJSONObject("operate_date").getString("order_id")
                        )
                ).between(Time.seconds(-1), Time.seconds(2))
                .process(new ProcessJoinFunction<String, String, String>() {
                    @Override
                    public void processElement(String info, String detail, Context ctx, Collector<String> out) throws Exception {
                        // 获取user_id和province_id
                        JSONObject infoJsonObject = JSON.parseObject(info).getJSONObject("operate_date");
                        Integer userId = infoJsonObject.getInteger("user_id");
                        Integer provinceId = infoJsonObject.getInteger("province_id");
                        // 将字段添加到Detail中
                        JSONObject detailJsonObject = JSON.parseObject(detail).getJSONObject("operate_date");
                        detailJsonObject.put("user_id", userId);
                        detailJsonObject.put("province_id", provinceId);
                        // 订单日期
                        String createTime = detailJsonObject.getString("create_time");
                        String curDate = createTime.substring(0, 10);
                        detailJsonObject.put("cur_date", curDate);
                        // 输出
                        out.collect(detailJsonObject.toJSONString());
                    }
                });
        return joinStream;
    }
}